/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.lvyh.lightframe.tracer.core.zipkin;

import com.lvyh.lightframe.tracer.core.SimpleSpan;
import com.lvyh.lightframe.tracer.core.SimpleSpanContext;
import com.lvyh.lightframe.tracer.core.common.TraceContants;
import com.lvyh.lightframe.tracer.core.util.NetUtils;
import com.lvyh.lightframe.tracer.core.util.NumberUtils;
import io.opentracing.tag.Tags;
import org.apache.commons.lang3.StringUtils;

import java.net.InetAddress;
import java.util.Map;

/**
 * SimpleTracer and Zipkin model transformation
 * In order to enable SimpleTracer's data to be parsed by Zipkin,
 * it is necessary to convert SimpleTracer's span model into Zipkin V2's data model.
 * <p>
 * Simplespan is based on the opentracing standard. However,
 * the opentracing standard does not specify which attributes a span model must have.
 * Therefore, the products based on this standard are not unified in the span model,
 * and most of them have some special properties based on their own scenes.
 */
public class ZipkinSpanConvertor {

    private static InetAddress localIpAddress = null;

    public ZipkinTraceSpan convertToZipkinSpan(SimpleSpan simpleSpan) {
        if (simpleSpan == null) {
            return null;
        }

        SimpleSpanContext context = simpleSpan.getSimpleSpanContext();

        ZipkinTraceSpan span = new ZipkinTraceSpan();
        /**
         * Zipkin did a lot of special processing in his model. For example, trace id needs to meet 16 or 32 bits.
         * If the length is not enough, the high bit will be supplemented with 0;
         * Therefore, when using simpletracer, it is reasonable that the traceid in the log is inconsistent with the traceid reported to Zipkin.
         */
        span.setTraceId(context.getTraceId());
        if (!StringUtils.isEmpty(context.getParentId())) {
            span.setParentId(formatSpanId(context.getParentId()));
        }
        span.setId(formatSpanId(context.getSpanId()));

        // timestamp & duration
        long start = simpleSpan.getStartTime() * 1000;
        long finish = simpleSpan.getEndTime() * 1000;
        span.setTimestamp(start);
        span.setDuration(finish - start);

        // kind
        Map<String, String> tags = simpleSpan.getTags();
        String kind = tags.get(Tags.SPAN_KIND.getKey());
        if (StringUtils.isNotBlank(kind) && kind.equals(Tags.SPAN_KIND_SERVER)) {
            span.setKind(ZipkinTraceSpan.Kind.SERVER.toString());
        } else {
            span.setKind(ZipkinTraceSpan.Kind.CLIENT.toString());
        }

        // Endpoint
        ZipkinTraceSpan.Endpoint endpoint1 = getZipkinEndpoint(simpleSpan);
        ZipkinTraceSpan.Endpoint endpoint2 = getZipkinRemoteEndpoint(simpleSpan);
        span.setLocalEndpoint(endpoint1);
        span.setRemoteEndpoint(endpoint2);

        // span name
        String operationName = simpleSpan.getOperationName();
        if (StringUtils.isNotBlank(operationName)) {
            span.setName(operationName);
        } else {
            span.setName("");
        }

        span.setTags(simpleSpan.getTags());
        return span;
    }

    public static String formatSpanId(String spanId) {
        long spanIdLong = spanIdToLong(spanId);
        return spanIdLong != 0L ? NumberUtils.toLowerHex(spanIdLong) : "0";
    }

    /**
     * In addition, if the parentid of the span reporting Zipkin is 0, Zipkin will not be set;
     * The ID of the first span of simpletracer starts from 0, which will lead to link construction failure.
     * If we try to change the starting ID, it will affect the whole model.
     * After verification and testing,
     * our final solution is to use the fnv64 hash algorithm with less conflict to convert the string type into long type to describe our spanid.
     */
    public static long spanIdToLong(String spanId) {
        return FNV64HashCode(spanId);
    }

    //"0"=af63ad4c86019caf
    public static long FNV64HashCode(String data) {
        //hash FNVHash64 : http://www.isthe.com/chongo/tech/comp/fnv/index.html#FNV-param
        long hash = 0xcbf29ce484222325L;
        for (int i = 0; i < data.length(); ++i) {
            char c = data.charAt(i);
            hash ^= c;
            hash *= 0x100000001b3L;
        }
        return hash;
    }

    private ZipkinTraceSpan.Endpoint getZipkinEndpoint(SimpleSpan span) {
        if (localIpAddress == null) {
            localIpAddress = NetUtils.getLocalAddress();
        }
        String appName = span.getTags().get(TraceContants.LOCAL_APP);
        return new ZipkinTraceSpan.Endpoint(appName, localIpAddress);
    }

    private ZipkinTraceSpan.Endpoint getZipkinRemoteEndpoint(SimpleSpan span) {
        if (localIpAddress == null) {
            localIpAddress = NetUtils.getLocalAddress();
        }
        String appName = span.getTags().get(TraceContants.REMOTE_APP);
        return new ZipkinTraceSpan.Endpoint(appName, localIpAddress);
    }


}
